package cn.getech.data.development.job.util

import java.io.{BufferedReader, InputStream, InputStreamReader}
import java.net.InetSocketAddress

import org.apache.spark.sql.types.{StringType, StructField, StructType}
import sun.net.ftp.FtpClient

import scala.collection.mutable.ListBuffer

object FtpUtil {
  /**
    * 连接FTP服务
    *
    * @param url      //IP地址
    * @param port     //端口号
    * @param username //用户名
    * @param password //密码
    * @return
    */
  def connectFtp(url:String,port:Int,username:String,password:String): FtpClient ={

    val address = new InetSocketAddress(url,port)
    val ftp = FtpClient.create()

    ftp.connect(address)
    ftp.login(username,password.toCharArray)
    ftp.setBinaryType()
    ftp.setConnectTimeout(120000)
    ftp
  }

  /**
    * 取ftp上的文件内容第一行，也就是column name
    *
    * @param ftpFile
    * @param ftp
    * @return
    */
  def downLoadColumn(ftpFile: String, ftp: FtpClient): String = {
    var column = ""
    val fs: InputStream = ftp.getFileStream(ftpFile)
    val br = new BufferedReader(new InputStreamReader(fs))
    column += br.readLine()
    br.close()
    fs.close()
    column
  }

  def schemaInfoOld(cloumn:String): Unit ={
    val columns: Array[String] = cloumn.split("  ")
    var col = "record_date"
    val structFieldList = new ListBuffer[StructField]()
    structFieldList += StructField(col,StringType,true)

    for ( i <- 1 until columns.length ){

      structFieldList += StructField(columns(i).toLowerCase,StringType,true)
      if(columns(i).toLowerCase == "type" ){
        col += (",`" +s"${columns(i).toLowerCase()}"+"`")
      }else{
        col += (","+columns(i).toLowerCase)
      }
    }

    val schema: StructType = StructType(structFieldList)
    structFieldList.clear()
    (schema,col,columns.length)
  }

  def getSchemaInfo(column:String):StructType = {

    val strings: Array[String] = column.split("\t")
    val structFieldList = new ListBuffer[StructField]()

    for (elem <- strings) {
      structFieldList += new StructField(elem,StringType,true)
    }

    val schema: StructType = StructType(structFieldList)
    structFieldList.clear()
    schema
  }

  def main(args: Array[String]): Unit = {
    val client: FtpClient = connectFtp("10.74.20.65",21,"ftpUser","ftpuser")
    val fs = client.getFileStream("/mergedModel.csv")
    val bufferReader = new BufferedReader(new InputStreamReader(fs))
    var ftpData = ""
    while(null != ftpData){
      Thread.sleep(10)
      ftpData = bufferReader.readLine()
      /*if( "" != ftpData && !"null".equals(ftpData) && null!=ftpData ){
      }*/
      if(null != ftpData){
        println(ftpData)
      }
    }
  }
}